﻿
Imports System.Threading
Imports std
Imports threadpoolCommon

Public Class threadpool
    Implements ithreadpool

    Private Const low_pri_rate As Double = 0
    Private Shared ReadOnly threadpoolLengthCountSampleRate As Double = min(counter.defaultSampleRate * 100, 0.01)
    Private Shared ReadOnly defaultThreadCount As Int64 = threadpoolCommon.defaultThreadCount() * (1 + low_pri_rate)
    <ThreadStatic()> Private Shared storage As map(Of String, pointer(Of Object))
    Private ReadOnly THREADPOOL_QUEUE_LENGTH As Int64
    Private ReadOnly THREADPOOL_WAIT_TICKS As Int64
    Private ReadOnly THREADPOOL_IDLE_ROUNDS As Int64
    Private ReadOnly q As qless(Of workInfo) = Nothing
    Private ReadOnly threads As vector(Of Thread) = Nothing
    Private ReadOnly are As AutoResetEvent = Nothing
    Private ReadOnly work As ParameterizedThreadStart = Nothing
    Private ReadOnly counterUpdaterId As Int64 = 0
    Private ReadOnly typename As String = Nothing
    Private _idleThreadCount As Int32 = 0
    Private stopsignal As Boolean = False
    Private _timelimitOperationEnabled As Boolean = True
    Private lastQueueLength As Int64 = 0
    Private _busyWaitingJob As Boolean = False

    Public Sub disableTimeLimitOperation() Implements ithreadpool.disableTimeLimitOperation
        _timelimitOperationEnabled = False
    End Sub

    Public Sub enableTimeLimitOperation() Implements ithreadpool.enableTimeLimitOperation
        _timelimitOperationEnabled = True
    End Sub

    Public Sub enableBusyWaitingJob()
        _busyWaitingJob = True
    End Sub

    Public Sub disableBusyWaitingJob()
        _busyWaitingJob = False
    End Sub

    Public Function busyWaitingJob() As Boolean
        Return _busyWaitingJob
    End Function

    Private Function timelimitOperationEnabled() As Boolean
        Return _timelimitOperationEnabled AndAlso timelimitOperation.enabled()
    End Function

    Public Shared Function threadStorage() As map(Of String, pointer(Of Object))
        If storage Is Nothing Then
            storage = New map(Of String, pointer(Of Object))()
        End If
        Return storage
    End Function

    Public Shared Sub store(ByVal key As String, ByVal obj As Object)
        threadStorage()(key) = New pointer(Of Object)(obj)
    End Sub

    Public Shared Function acquire(ByVal key As String) As Object
        Dim it As map(Of String, pointer(Of Object)).iterator = Nothing
        it = threadStorage().find(key)
        If it = threadStorage().end() Then
            Return Nothing
        Else
            Return +((+it).second)
        End If
    End Function

    Public Function queueLength() As Int64
        Return q.size()
    End Function

    Public Function idleThreadCount() As Int32
        Return atomic.read(_idleThreadCount)
    End Function

    Public Function idle() As Boolean
        Return queueLength() = 0 AndAlso idleThreadCount() = threadCount()
    End Function

    Private Sub wait()
        Interlocked.Increment(_idleThreadCount)
        are.WaitOne()
        Interlocked.Decrement(_idleThreadCount)
    End Sub

    Private Sub waitJob(ByVal busyWaiting As Boolean)
        If busyWaiting Then
            If try_wait_when(AddressOf q.empty) Then
                wait()
            End If
        Else
            wait()
        End If
    End Sub

    Private Sub setJob()
        If idleThreadCount() > 0 Then
            are.Set()
        End If
    End Sub

    Public Sub insertJob(ByVal obj As Object) Implements ithreadpool.insertJob
        insertJob(work, obj)
    End Sub

    Public Overridable Sub insertJob(ByVal work As ParameterizedThreadStart, ByVal obj As Object) _
                                    Implements ithreadpool.insertJob
        If stopsignal Then
            If envs.perftrace() Then
                raiseError("should not insertJob after stop, ", callstack())
            End If
            Return
        End If

        q.push(New workInfo(work, obj))
        setJob()
    End Sub

    Private Sub workthread()
        While True
            Dim obj As workInfo = Nothing
            obj = q.pop()
            If obj Is Nothing Then
                If needtrace Then
                    Dim startticks As Int64 = 0
                    startticks = Now.ticks()
                    waitJob(busyWaitingJob())
                    counter.recordTimeTicks(THREADPOOL_WAIT_TICKS, startticks)
                    counter.increase(THREADPOOL_IDLE_ROUNDS)
                Else
                    waitJob(busyWaitingJob())
                End If
            Else
                workOn(obj, timelimitOperationEnabled:=timelimitOperationEnabled())
            End If
        End While
    End Sub

    Private Function typedname(ByVal s As String) As String
        Return strcat(typename, s)
    End Function

    Private Function upper_typedname(ByVal s As String) As String
        Return strtoupper(typedname(s))
    End Function

    Public Property threadCount() As Int64 Implements ithreadpool.threadCount
        Get
            Return threads.size()
        End Get
        Set(ByVal value As Int64)
            If value <= 0 Then
                value = defaultThreadCount
            End If
            If threads.size() > value Then
                [stop]()
            End If
            While threads.size() < value
                threads.push_back(startThread(AddressOf workthread))
                Dim t As Thread = Nothing
                t = +(threads.rbegin())
                If low_pri_rate > 0 Then
                    If threads.size() <= value / (1 + low_pri_rate) Then
                        t.Name() = upper_typedname("_workthread_hi_pri")
                        t.Priority() = ThreadPriority.Highest
                    Else
                        t.Name() = upper_typedname("_workthread_low_pri")
                        t.Priority() = ThreadPriority.Lowest
                    End If
                Else
                    t.Name() = upper_typedname("_workthread")
                End If
            End While
        End Set
    End Property

    Public Sub New(ByVal workthread As ParameterizedThreadStart, Optional ByVal threadCount As Int64 = 0)
        typename = Me.GetType().Name()
        If needtrace Then
            THREADPOOL_WAIT_TICKS = counter.registerAverageAndLastAverage(upper_typedname("_WAIT_TICKS"))
            THREADPOOL_IDLE_ROUNDS = counter.registerRateAndLastRate(upper_typedname("_IDLE_ROUNDS"))
        End If
        THREADPOOL_QUEUE_LENGTH = counter.registerAverageAndLastAverage(upper_typedname("_QUEUE_LENGTH"),
                                                                        sampleRate:=threadpoolLengthCountSampleRate)
        q = New qless(Of workInfo)()
        threads = New vector(Of Thread)()
        are = New AutoResetEvent(False)
        AddHandler applicationShutdown, AddressOf [stop]
        counterUpdaterId = queueInBackendThread(AddressOf regularCounterUpdater)
        work = workthread
        Me.threadCount() = threadCount
        stopsignal = False
    End Sub

    Public Sub New(ByVal threadCount As Int64)
        Me.New(Nothing, threadCount)
    End Sub

    Public Sub New()
        Me.New(defaultThreadCount)
    End Sub

    Public Function stopping() As Boolean Implements ithreadpool.stopping
        Return stopsignal
    End Function

    Public Sub [stop](Optional ByVal stopWaitSeconds As Int64 = defaultStopWaitSeconds) Implements ithreadpool.stop
        threadpoolCommon.stop(threads, stopWaitSeconds,
                              Sub()
                                  stopsignal = True
                                  If threadCount() > 0 Then
                                      While Not idle() AndAlso stopWaitSeconds > 0
                                          sleepSeconds(1)
                                          stopWaitSeconds -= 1
                                      End While
                                  End If
                              End Sub,
                              Sub()
                                  threads.clear()
                                  q.clear()
                              End Sub)
    End Sub

    Public Shared ReadOnly [global] As threadpool = Nothing

    Shared Sub New()
        [global] = New threadpool()
    End Sub

    Public Shared Sub register(ByVal resolver As resolver)
        resolver.register(Of ithreadpool)([global])
        resolver.register([global])
    End Sub

    Public Shared Sub register()
        register(resolver.global)
    End Sub

    Private Sub regularCounterUpdater()
        counter.increase(THREADPOOL_QUEUE_LENGTH, AddressOf queueLength)
    End Sub

    Protected Overrides Sub Finalize()
        removeFromBackendThread(counterUpdaterId)
        are.Close()
        GC.KeepAlive(are)
        MyBase.Finalize()
    End Sub
End Class

Public Class adaptingThreadPool
    Inherits threadpool

    Public adaptingLimit As Int64 = 32
    Public Const threadUpperLimit As Int64 = 512

    Public Overrides Sub insertJob(ByVal work As ParameterizedThreadStart, ByVal obj As Object)
        MyBase.insertJob(work, obj)
        If queueLength() > adaptingLimit AndAlso threadCount() < threadUpperLimit Then
            threadCount() >>= 1
        End If
    End Sub
End Class

Public Class lazyThreadPool
    Inherits threadpool

    Public Shared Shadows ReadOnly [global] As lazyThreadPool = Nothing

    Shared Sub New()
        [global] = New lazyThreadPool()
    End Sub

    Public Shared Shadows Sub register(ByVal resolver As resolver)
        resolver.register([global])
    End Sub

    Public Shared Shadows Sub register()
        register(resolver.global)
    End Sub

    Private Sub initial()
        Me.threadCount() = Environment.ProcessorCount() << 4
        disableTimeLimitOperation()
    End Sub

    Public Sub New(ByVal workthread As ParameterizedThreadStart, Optional ByVal threadCount As Int64 = 0)
        MyBase.New(workthread, threadCount)
    End Sub

    Public Sub New(ByVal threadCount As Int64)
        MyBase.New(threadCount)
    End Sub

    Public Sub New()
        MyBase.New()
    End Sub
End Class
